package org.apache.lucene.ocean.log;

import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.lucene.ocean.LogDirectory;
import org.apache.lucene.ocean.log.RawLogFile.CRCException;
import org.apache.lucene.ocean.log.RawLogFile.FileStreamRecord;
import org.apache.lucene.ocean.log.RawLogFile.StreamData;
import org.apache.lucene.ocean.log.RawLogFile.StreamRecord;
import org.apache.lucene.ocean.util.OceanRandomAccessFile;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Log file.  Contains all record headers for loading of the actual record.
 *
 */
// TODO: RecordHeader in memory needs to only have the pointer to the record data
public class LogFile implements Comparable<LogFile> {
  final static Logger LOG = LoggerFactory.getLogger(LogFile.class);
  List<RecordHeader> recordHeaders; // sorted by id
  private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  private RawLogFile rawLogFile;
  private Long id;
  
  LogFile(Long id) {
    this.id = id;
  }
  
  /**
   * 
   * @param id Id of the log file
   * @param file File name of the log file
   * @param logDirectory LogDirectory to use
   * @throws IOException
   */
  public LogFile(Long id, String file, LogDirectory logDirectory) throws IOException {
    this.id = id;
    rawLogFile = new RawLogFile(file, logDirectory);
    recordHeaders = rawLogFile.loadRecordHeaders();
    if (recordHeaders.size() > 0) {
      // check to make sure last record was completely written
      // by checking the crc32
      RecordHeader last = recordHeaders.get(recordHeaders.size()-1);
      try {
        checkCrc(last.id);
      } catch (CRCException crcException) {
        handleLastHeaderFailure();
      } catch (EOFException eofException) {
        handleLastHeaderFailure();
      }
    }
  }
  
  private void handleLastHeaderFailure() {
    // last one did not make it, remove it
    RecordHeader last = recordHeaders.get(recordHeaders.size()-1);
    LOG.info("removing last record "+last.id+" as it did not make it");
    recordHeaders.remove(recordHeaders.size()-1);
    // reset to position after previous known header
    rawLogFile.resetWritePosition(recordHeaders);
  }
  
  /**
   * Get the name of the file
   * @return
   */
  public String getName() {
    return rawLogFile.getName();
  }
  
  /**
   * Get the number of records
   * @return
   */
  public int getNumRecords() {
    return recordHeaders.size();
  }
  
  /**
   * Delete the log file
   * @return If the log file was deleted
   * @throws IOException
   */
  public boolean delete() throws IOException {
    close();
    return rawLogFile.delete();
  }
  
  public void close() throws IOException {
    rawLogFile.close();
  }
  
  public long getPreviousId(long id) {
    long pos = getPosition(id);
    if (pos == 0) return -1;
    return pos-1;
  }
  
  /**
   * Compare the ids of the LogFiles
   */
  public int compareTo(LogFile other) {
    return (id < other.id ? -1 : (id == other.id ? 0 : 1));
  }

  private void add(RecordHeader recordHeader) {
    recordHeaders.add(recordHeader);
  }
  
  /**
   * The size in bytes of the log file
   * @return
   * @throws IOException
   */
  public long size() throws IOException {
    return rawLogFile.size();
  }

  public int getPosition(long id) {
    RecordHeader h = new RecordHeader();
    h.id = id;
    int pos = Collections.binarySearch(recordHeaders, h);
    if (pos < 0)
      pos = -1 - pos;
    return pos;
  }
  
  public int getRawPosition(long id) {
    RecordHeader h = new RecordHeader();
    h.id = id;
    return Collections.binarySearch(recordHeaders, h);
  }
  
  /**
   * Check the crc values of the record
   * @param id The id of the record to check
   * @throws IOException
   */
  public void checkCrc(Long id) throws IOException {
    RecordIterator iterator = getRecordIterator(id);
    if (iterator.hasNext()) {
      Record record = iterator.next();
      StreamRecord streamRecord = record.getStreamRecord();
      StreamData docData = streamRecord.getDocuments();
      if (docData != null) docData.verifyCrc();
      StreamData otherData = streamRecord.getOther();
      if (otherData != null) otherData.verifyCrc();
    } else {
      throw new IOException("id: "+id+" does not exist");
    }
  }
  
  public RecordHeader get(long id) {
    if (recordHeaders.size() > 0) {
      int pos = getPosition(id);
      if (pos >= recordHeaders.size()) return null;
      return recordHeaders.get(pos);
    } else return null;
  }
  /**
  public boolean delete(long id) throws IOException {
    lock.writeLock().lock();
    try {
      RecordHeader recordHeader = get(id);
      boolean deleted = rawLogFile.delete(recordHeader);
      if (deleted) {
        int pos = getPosition(id);
        recordHeaders.remove(pos);
      }
      return deleted;
    } finally {
      lock.writeLock().unlock();
    }
  }
  **/
  public long getId() {
    return id;
  }
  
  /**
   * Gets the minimum id of the log file
   * @return
   */
  public long getMinId() {
    if (recordHeaders.size() == 0)
      return -1;
    else
      return recordHeaders.get(0).id;
  }
  
  /**
   * If the log file contains the transaction id based on the min max range
   * @param id
   * @return
   */
  public boolean containsId(Long id) {
    if (recordHeaders.size() == 0) return false;
    return id <= getMaxId() && id >= getMinId();
  }
  
  /**
   * Get the maximum id of the record headers
   * @return
   */
  public Long getMaxId() {
    if (recordHeaders.size() == 0) return null;
    return recordHeaders.get(recordHeaders.size() - 1).id;
  }
  
  /**
   * Get the record headers 
   * @param from Snapshot id to start from
   * @param num Maximum number of headers to return
   * @return
   */
  public List<RecordHeader> getRecordHeaders(long from, int num) {
    int pos = getPosition(from);
    int count = 0;
    List<RecordHeader> recordHeaders = new ArrayList<RecordHeader>(num);
    while (count < num) {
      RecordHeader recordHeader = recordHeaders.get(pos);
      recordHeaders.add(recordHeader);
      count++;
    }
    return recordHeaders;
  }

  public void commit(RecordHeader recordHeader) {
    // TODO: commit
  }
  
  /**
   * Writes the record to the log file.
   * @param id
   * @param recordData
   * @return
   * @throws IOException
   */
  public RecordHeader writeRecord(Long id, RecordData recordData) throws IOException {
    lock.writeLock().lock();
    try {
      RecordHeader recordHeader = rawLogFile.write(id, recordData.docType, recordData.docBytes, recordData.otherType, recordData.otherBytes);
      add(recordHeader);
      return recordHeader;
    } finally {
      lock.writeLock().unlock();
    }
  }
  
  // TODO: needs parameter that defines type
  //public ListIterator getListIterator() {
  //  return recordHeaders.listIterator();
  //}
  
  /**
   * Get the record for the id
   */
  public Record getRecord(long id) throws IOException {
    RecordIterator iterator = getRecordIterator(id);
    if (iterator.hasNext()) {
      return iterator.next();
    }
    return null;
  }
  
  /**
   * Get the RecordIterator
   * @param snapshotId Snapshot id to start from
   * @return
   * @throws IOException
   */
  public RecordIterator getRecordIterator(Long snapshotId) throws IOException {
    OceanRandomAccessFile input = rawLogFile.openInput();
    return new RecordIterator(snapshotId, input);
  }
  
  /**
   * Iterates over the records stored in an OceanRandomAccessFile.
   *
   */
  public class RecordIterator {
    private OceanRandomAccessFile input;
    int pos;

    public RecordIterator(Long snapshotId, OceanRandomAccessFile input) throws IOException {
      this.input = input;
      if (snapshotId == null) {
        pos = 0;
      } else {
        pos = getPosition(snapshotId);
      }
    }

    public Record next() throws IOException {
      RecordHeader recordHeader = recordHeaders.get(pos);
      if (LOG.isDebugEnabled()) LOG.debug("recordHeader: "+recordHeader);
      pos++;
      return getRecord(recordHeader);
    }

    private Record getRecord(RecordHeader recordHeader) throws IOException {
      FileStreamRecord fileStreamRecord = rawLogFile.readRecordData(recordHeader, input);
      Record record = new Record(recordHeader.id, recordHeader, fileStreamRecord);
      return record;
    }

    public boolean hasNext() {
      return pos < recordHeaders.size();
    }

    public void close() throws IOException {
      input.close();
    }
  }
  
  /**
   * Represents a record.  The actual data is loaded using the StreamRecord.
   *
   */
  public static class Record {
    private long id;
    private StreamRecord streamRecord;
    private RecordHeader recordHeader;

    public Record(long id, RecordHeader recordHeader, StreamRecord streamRecord) {
      this.id = id;
      this.recordHeader = recordHeader;
      this.streamRecord = streamRecord;
    }
    
    public RecordHeader getRecordHeader() {
      return recordHeader;
    }
    
    public Long getId() {
      return id;
    }

    public StreamRecord getStreamRecord() {
      return streamRecord;
    }
  }
}
